package com.atguigu.flink.chapter04;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/19 14:54
 */
public class Flink01_Runtime_Parallelism {
    public static void main(String[] args) throws Exception {

        // 创建执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 为了idea运行直接可以看到 webui，使用如下方式：
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        // 全局设置并行度
        env.setParallelism(1);

        env.disableOperatorChaining(); // 全局禁用操作链

        env
                .socketTextStream("localhost", 9999)
//                .socketTextStream("hadoop1", 9999)
                .flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String value, Collector<String> out) throws Exception {
                        String[] words = value.split(" ");
                        for (String word : words) {
                            out.collect(word);
                        }
                    }
                })
//                .disableChaining()  // 当前算子不加入任何链条
                .map(new MapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String value) throws Exception {
                        return Tuple2.of(value, 1L);
                    }
                })
                //.setParallelism(4)
//                .startNewChain()    // 以当前算子开始，重新开链条
                .keyBy(0)
                .sum(1)
                .print();

        // 执行
        env.execute();
    }
}

/**
 * 并行度的优先级：
 * 算子指定(代码) > env全局指定（代码） > 提交参数 > 配置文件
 * <p>
 * 并行度与slot数量的关系：
 * （如果不是Yarn模式，不会动态申请资源）如果并行度 > 所有的slot数 ，程序会一直处于 create状态，等待足够的资源，才运行
 * <p>
 * 如果程序idea直接运行，并且没有指定并行度，那么默认使用 本机器的线程数 为并行度
 * <p>
 * Task与 subtask： 不同算子的 子任务（subtask） 经过一定的优化，串在一起，形成一个 新的subtask，叫做task
 * <p>
 * sum算子   print算子 ， 并行度都是 3
 * ⭕          ⭕
 * ⭕          ⭕
 * ⭕          ⭕
 * <p>
 * sum算子 跟 print算子 满足某种不可描述的关系，可以串在一起
 * （⭕  ⭕） -> 新的 subtask -> 对 TaskManager来讲，就是一个 Task
 * （⭕  ⭕） -> 新的 subtask -> 对 TaskManager来讲，就是一个 Task
 * （⭕  ⭕） -> 新的 subtask -> 对 TaskManager来讲，就是一个 Task
 */


/**
 * Task与 subtask： 不同算子的 子任务（subtask） 经过一定的优化，串在一起，形成一个 新的subtask，叫做task
 *
 *  sum算子   print算子 ， 并行度都是 3
 *   ⭕          ⭕
 *   ⭕          ⭕
 *   ⭕          ⭕
 *
 *  sum算子 跟 print算子 满足某种不可描述的关系，可以串在一起
 *      （⭕  ⭕） -> 新的 subtask -> 对 TaskManager来讲，就是一个 Task
 *      （⭕  ⭕） -> 新的 subtask -> 对 TaskManager来讲，就是一个 Task
 *      （⭕  ⭕） -> 新的 subtask -> 对 TaskManager来讲，就是一个 Task
 *
 */

/**
 * 操作链相关API
 *  1） 算子.startNewChain()   =》 与前面切开
 *  2） 算子.disableChaining() =》 与前后都切开
 *  3） env.disableOperatorChaining() =》 全局都不串
 */